package com.whoami.dataplatform.tag_combine.old;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.BytesWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.input.SequenceFileAsBinaryInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;

import java.io.IOException;

/**
 *
 * @author tzp
 * @since 2019/7/31
 */
@SuppressWarnings("Duplicates")
public class TidbitsCombine2 extends Configured implements Tool {

    @Override
    public int run(String[] args) throws Exception {
        Configuration conf = getConf();
        String date = args[0];//20190723
        String input = args[1];
        String output = args[2];//outputbase


        Job job = Job.getInstance(conf, this.getClass().getSimpleName() + " " + date);
        job.setJarByClass(this.getClass());

        //输入
        job.setInputFormatClass(SequenceFileAsBinaryInputFormat.class);
        FileInputFormat.addInputPath(job, new Path(input));

        //mapper
        job.setMapperClass(FilterMapper.class);

        //reducer
        job.setNumReduceTasks(0);

        //输出
        job.setOutputFormatClass(TextOutputFormat.class);

        FileOutputFormat.setOutputPath(job, new Path(output));

        boolean result = job.waitForCompletion(true);

        return result ? 0 : 1;
    }

    public static class FilterMapper extends Mapper<BytesWritable, BytesWritable, NullWritable, Text> {
        Text outValue = new Text();

        @Override
        protected void map(BytesWritable key, BytesWritable value, Context context) throws IOException, InterruptedException {
            outValue.set(value.getBytes(), 0, value.getLength());
            context.write(NullWritable.get(), outValue);
        }
    }

    public static void main(String[] args) throws Exception {
        System.exit(ToolRunner.run(new Configuration(), new TidbitsCombine2(), args));
    }
}
